//package com.deer.wms.base.system.web.rocketmq;
//
//
//import com.deer.wms.base.system.util.JedisServiceUtil;
//import lombok.extern.slf4j.Slf4j;
//import org.apache.rocketmq.client.consumer.DefaultMQPullConsumer;
//import org.apache.rocketmq.client.consumer.PullResult;
//import org.apache.rocketmq.client.exception.MQClientException;
//import org.apache.rocketmq.common.message.MessageExt;
//import org.apache.rocketmq.common.message.MessageQueue;
//import org.springframework.beans.factory.annotation.Autowired;
//import org.springframework.beans.factory.annotation.Value;
//import org.springframework.context.annotation.Bean;
//import org.springframework.stereotype.Service;
//import javax.annotation.PostConstruct;
//import javax.annotation.PreDestroy;
//import java.text.SimpleDateFormat;
//import java.util.*;
//
///**
// *
// */
//@Service
//@Slf4j
//public class PullConsumer implements Runnable {
//
//    @Autowired
//    private JedisServiceUtil jedis;
//
//    /**
//     * 消费者
//     */
//    @Bean
//    public DefaultMQPullConsumer DefaultMQPullConsumer() {
//        return new DefaultMQPullConsumer("SpringBootRocketMqGroup");
//    }
//
//    @Value("${apache.rocketmq.consumer.pushConsumer}")
//    private String pushConsumer;
//
//    /**
//     * NameServer 地址
//     */
//    @Value("${apache.rocketmq.namesrvAddr}")
//    private String namesrvAddr;
//
//    @Autowired
//    private DefaultMQPullConsumer consumer;
//
//    public static int task;
//
//    private static PullConsumer pullConsumer = new PullConsumer();
//
//    public PullConsumer() {
//        task = 0;
//    }
//
//    public static PullConsumer getInstance() {
//        return pullConsumer;
//    }
//
//
//    //保存上一次消费的消息位置
//    private static final Map offsetTable = new HashMap();
//
//    private static final List<MessageExt> lMessageExt = new ArrayList<>();
//
//    @PostConstruct
//    public void init() {
//        task = 0;
//        pullConsumer = this;
//        pullConsumer.jedis = this.jedis;
//        consumer.setNamesrvAddr(namesrvAddr);
//        try {
//            consumer.setConsumerPullTimeoutMillis(3000);
//            consumer.start();
//        } catch (MQClientException e) {
//            log.error(e.getStackTrace().toString());
//        }
//        consumer.setBrokerSuspendMaxTimeMillis(10);
//    }
//
//    /**
//     * 拉取消息,并把消息放到消息的list中
//     * @author 申紫微
//     * @date 14:13 2020/9/17
//     * @params
//     * @return void
//     **/
//    public void dealMsg() {
//        //实例化pullConsumer
//        //拉取PullConsumerTopicTest topic下的所有消息队列
//        Set<MessageQueue> mqs = null;
//        try {
//            mqs = consumer.fetchSubscribeMessageQueues("inWarehouseTopic-test");
//        } catch (MQClientException e) {
//            log.error(e.getErrorMessage());
//        }
//        task = 1;
//        //遍历消息队列
//        if(null == mqs || mqs.size() ==0){
//            return;
//        }
//        for (MessageQueue mq : mqs) {
//            log.info("Consume from the queue:{}",mq);
//            SINGLE_MQ:
//            try {
//                System.out.println("begin:" + new SimpleDateFormat("yyyymmdd HH:mm:ss:SSS").format(new Date()));
//                PullResult pullResult = consumer.pullBlockIfNotFound(mq, null, getMessageQueueOffset(), 32);
//                System.out.println(pullResult);
//                System.out.println("end:" + new SimpleDateFormat("yyyymmdd HH:mm:ss:SSS").format(new Date()));
//                //消费完消息之后,设置上次消费消息的下标
////                putMessageQueueOffset(mq, pullResult.getNextBeginOffset());
//                switch (pullResult.getPullStatus()) {
//                    //根据结果状态，如果找到消息，批量消费消息
//                    case FOUND:
//                        List<MessageExt> messageExtList = pullResult.getMsgFoundList();
////                        for (MessageExt m : messageExtList) {
////                            System.out.println(m.getMsgId() + ":" + m.getQueueId() + ":" + new String(m.getBody()));
////                        }
//                        // 放消息到list中
//                        lMessageExt.addAll(messageExtList);
//                        break;
//                    case NO_MATCHED_MSG:
//                        break;
//                    case NO_NEW_MSG:
//                        break SINGLE_MQ;
//                    case OFFSET_ILLEGAL:
//                        break;
//                    default:
//                        break;
//                }
//            } catch (Exception e) {
//                log.error(e.getStackTrace().toString());
//            }
//        }
//        task = 0;
//    }
//
//    //保存上次消费的消息下标，这里使用了一个全局HashMap来保存
//    //修改为通过redis来存储下标
//    public static void putMessageQueueOffset(MessageQueue mq, long offset) {
////        offsetTable.put(mq, offset);
//        boolean mqOffset = pullConsumer.jedis.set("mqOffset", String.valueOf(offset));
//    }
//
//    //获取上次消费的消息的下表
//    public static long getMessageQueueOffset() {
////        Long offset = (Long) offsetTable.get(mq);
//        String offset = pullConsumer.jedis.get("mqOffset");
//        if (offset != null) {
//            return Long.parseLong(offset);
//        }
//        return 0;
//    }
//
//    public MessageExt getMessageExt() {
//        if (lMessageExt.size() > 0) {
//            MessageExt messageExt = lMessageExt.remove(0);
//            return messageExt;
//        } else {
//            return null;
//        }
//    }
//
//    @PreDestroy
//    public void preDestory() {
//        log.info("程序关闭......关闭消息队列消费者");
//        consumer.shutdown();
//    }
//
//    @Override
//    public void run() {
//        log.info("......执行消息拉取......");
//        Set<MessageQueue> mqs = null;
//        try {
//            mqs = consumer.fetchSubscribeMessageQueues("inWarehouseTopic-test");
//        } catch (MQClientException e) {
//            log.error(e.getStackTrace().toString());
//        }
//        //遍历消息队列
//        for (MessageQueue mq : mqs) {
//            System.err.println("Consume from the queue: " + mq);
//            SINGLE_MQ:
//            //
//            while (true) {
//                try {
//                    //设置上次消费消息下标
//                    PullResult pullResult = consumer.pullBlockIfNotFound(mq, "pull", getMessageQueueOffset(), 32);
//                    System.out.println(pullResult);
//                    putMessageQueueOffset(mq, pullResult.getNextBeginOffset());
//                    switch (pullResult.getPullStatus()) {
//                        //根据结果状态，如果找到消息，批量消费消息
//                        case FOUND:
//                            List<MessageExt> messageExtList = pullResult.getMsgFoundList();
////                            for (MessageExt m : messageExtList) {
////                                log.info("messageBody:{}", new String(m.getBody()));
////                            }
//                            lMessageExt.addAll(messageExtList);
//                            break;
//                        case NO_MATCHED_MSG:
//                            break;
//                        case NO_NEW_MSG:
//                            break SINGLE_MQ;
//                        case OFFSET_ILLEGAL:
//                            break;
//                        default:
//                            break;
//                    }
//                } catch (Exception e) {
//                    log.error(e.getStackTrace().toString());
//                }
//            }
//        }
//    }
//
//    //    @Override
//    public void run(String... args) throws Exception {
//        while (true) {
//            if (task == 0) {
//                dealMsg();
//            }
//            System.out.println("task=" + task);
//            Thread.sleep(5000);
//        }
//    }
//}